#ifndef _WORKER_WITH_QUEUE_H_
#define _WORKER_WITH_QUEUE_H_
#include <thread>
#include <atomic>
#include <queue>
#include <deque>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <functional>
#include <algorithm>

template<typename T> class SafeQueue;

class WorkerNotQueue
{
public:
    WorkerNotQueue(std::function<void()> func= nullptr)
        : _func(func)
    {

    }

    virtual ~WorkerNotQueue() {
        this->Uninit();
    }

    virtual void Run() { 
        // do nothing.
    }

    bool Init() {
        _isExit = false;
        _thread = std::thread([this]{ this->DoRun(); });
        return true;
    }

    void Uninit() { 
        _isExit = true;  
        if (_thread.joinable()) {
            _thread.join();
        }
    }

private:
    virtual void DoRun() {
        while (!_isExit) {
            Run();
            if (_func != nullptr) {
                _func();
            }
        }
    }

private:
    std::thread _thread;
    std::atomic<bool> _isExit { false };
    std::function<void()> _func = nullptr;
};


/**
 * @brief 带队列的工作者, 实现生产与消费模型
 * 
 */
template<typename T>
class WorkerWithQueue
{
public:
    WorkerWithQueue(std::function<void(T &data)> func = nullptr, const int &queueTimeoutMs = 1)
        : _func(func), _timeout(queueTimeoutMs)
    {
    }

    ~WorkerWithQueue() {
        this->Uninit();
    }

    virtual void Run(const T &data) { 
        // do nothing.
    }

    void PushBack(const T &data) {
        _queue.push(data);
    }

    bool Init(const int &threadCnt=1) {
        _isExit = false;
        for (int i=0; i<threadCnt; ++i) {
            _threads.push_back(std::thread([this]{ this->DoRun(); }));
        }
        return true;
    }
       
    void Uninit() { 
        _isExit = true;  
        _queue.NotifyAll();
        for (std::thread &th : _threads) {
            if (th.joinable()) {
                th.join();
            }
        }
    }

private:
    virtual void DoRun() {
        while (!_isExit) {
            T data; 
            if (_queue.wait_for_and_pop(data, _timeout)) {
                Run(data);
                if (_func != nullptr) {
                    _func(data);
                }
            }
        }
        
    }

private:
    SafeQueue<T> _queue;
    std::vector<std::thread> _threads;
    std::atomic<bool> _isExit { false };
    std::function<void(T &data)> _func;
    int _timeout = 1;
};


template<typename T>
class SafeQueue
{
public:
    SafeQueue() {}
    ~SafeQueue() { _isExit = true; m_cond.notify_all(); }

    void push(T new_data) {
        std::lock_guard<std::mutex> lk(m_mut); // 1.全局加锁
        m_dataQueue.push(std::move(new_data)); // 2.push时独占锁
        m_cond.notify_one();
    }

    bool wait_for_and_pop(T& val, const int &ms=1) {
        std::unique_lock<std::mutex> ulk(m_mut); // 3.全局加锁
        if (ms <= 0) {
//            if (!m_cond.wait(ulk, [this]() { return !m_dataQueue.empty() || _isExit; })) {
//                return false;
//            }
            return false;
        } else {
            if (!m_cond.wait_for(ulk, std::chrono::milliseconds(ms), [this]()
                                 { return !m_dataQueue.empty() || _isExit; }))
            {
                return false;
            }
        }
        if (_isExit) { return false; }
        val=std::move(m_dataQueue.front());
        m_dataQueue.pop();
        return true;
    }

    bool empty() {
        std::lock_guard<std::mutex> lk(m_mut);
        return m_dataQueue.empty();
    }

    int count() {
        std::lock_guard<std::mutex> lk(m_mut);
        return m_dataQueue.size();
    }

    void clear() {
        std::lock_guard<std::mutex> lk(m_mut);
        while (!m_dataQueue.empty()) {
            m_dataQueue.pop();
        }
    }

    void NotifyAll() { m_cond.notify_all(); }

private:
    std::queue<T> m_dataQueue;
    std::mutex m_mut;
    std::condition_variable m_cond;
    std::atomic<bool> _isExit { false };
};

template<typename T>
class SafeDeque
{
public:
    SafeDeque() {}
    ~SafeDeque() { _isExit = true; m_cond.notify_all(); }

    void push(T new_data) {
        std::lock_guard<std::mutex> lk(m_mut); // 1.全局加锁
        m_dataQueue.push_back(std::move(new_data)); // 2.push时独占锁
        m_cond.notify_one();
    }

    bool wait_for_and_pop(T& val, const int &ms=1) {
        std::unique_lock<std::mutex> ulk(m_mut); // 3.全局加锁
        if (ms <= 0) {
//            if (!m_cond.wait(ulk, [this]() { return !m_dataQueue.empty() || _isExit; })) {
//                return false;
//            }
            return false;
        } else {
            if (!m_cond.wait_for(ulk, std::chrono::milliseconds(ms), [this]()
                                 { return !m_dataQueue.empty() || _isExit; }))
            {
                return false;
            }
        }
        if (_isExit) { return false; }
        val=std::move(m_dataQueue.front());
        m_dataQueue.pop_front();
        return true;
    }

    bool empty() {
        std::lock_guard<std::mutex> lk(m_mut);
        return m_dataQueue.empty();
    }

    int count() {
        std::lock_guard<std::mutex> lk(m_mut);
        return m_dataQueue.size();
    }

    void clear() {
        std::lock_guard<std::mutex> lk(m_mut);
        m_dataQueue.clear();
    }

    // 存在则用最新的数据替换旧的值，位置保持不变
    void pushExistReplace(T new_data) {
        std::lock_guard<std::mutex> lk(m_mut); // 1.全局加锁
        auto iter = std::find_if(m_dataQueue.begin(), m_dataQueue.end(), [this, &new_data](T& item) {
            return item == new_data;
        });
        if (iter == m_dataQueue.end()) {
            m_dataQueue.push_back(std::move(new_data)); // 2.push时独占锁
        } else {
            *iter = new_data;
        }
        m_cond.notify_one();
    }

    void NotifyAll() { m_cond.notify_all(); }

private:
    std::deque<T> m_dataQueue;
    std::mutex m_mut;
    std::condition_variable m_cond;
    std::atomic<bool> _isExit { false };
};

#endif // _WORKER_WITH_QUEUE_H_
